はじめに

先日、AWS Summit San Francisco 2022にて、Amazon AthenaのFederated Queryを経由してデータにアクセスできる10種類のデータソースコネクタをサポートが新たに追加されました。ザックリ言うと、Athenaから以下のデータソースにアクセスできるようになったということです！ 本日は、中でも注目度の高いBigQueryへのクエリについて試してみます。

• SAP HANA (Express Edition)
• Teradata
• Cloudera
• Hortonworks
• Snowflake
• Microsoft SQL Server
• Oracle
• Azure Data Lake Storage (ADLS) Gen2
• Azure Synapse
• Google BigQuery

Amazon Athenaのデータソースコネクタが動作するVPC/Subnet環境

データソースコネクタの実態は、VPC Lambdaです。そのため、プライベートIPアドレスしか持たないVPC Lambdaから、後述するS3やSecrets ManagerのようなAWSのパブリックサービスやBigQueryに接続するためには、NATゲートウェイやVPC エンドポイントなどを経由してアクセスできる必要があります。

今回の動作検証では、以下のようなVPC/Subnet環境を用意しています。

• NATゲートウェイは、パブリックサービスであるSecrets ManagerやBigQueryに接続するため
• S3は、一時的なデータの退避（Spill Off）するためのS3バケットへのアクセス
• Secrets Managerは、保存したBigQueryの秘密鍵へアクセスするため

以降では、早速試したみたいと思います！

BigQueryの認証情報をSecrets Managerに作成

Google Cloudのサービスアカウントの秘密鍵を取得

事前にBigQueryにアクセスするためのGoogle Cloudのサービスアカウントを用意して、そのアカウントの秘密鍵を取得します。

秘密鍵は、jsonフォーマットでファイルをダウンロードします。

Secrets Managerに登録

Step1: [Other type of secret]を選択、次に[Plaintext]を選択して先程の秘密鍵の中身をコピーして、[Next]ボタンに進みます。

Step2: Secret nameを指定して、[Next]ボタンに進みます。

Step3: ストアした秘密鍵を自動的にローテーションできないため、変更することなく[Next]ボタンに進みます。

Step4: 変更することなく[Store]ボタンを押すとSecretの作成完了です。

デプロイが終わりますと、SecretのARNが割り当てられます。SecretのARNは、作成したSecretはSecretにアクセスするためのキーとして、以降で利用します。

BigQueryデータソースコネクタを試してみる！

Serverless Application Repositoryから検索

利用に先立って、Serverless Application Repository（東京リージョン）の画面からデータソースコネクタを検索してクリックします。

BigQueryデータソースコネクタのデプロイ

データソースコネクタの実態は、Lambda Functionなので、Lambdaの[Review, configure and deploy] 画面が表示されます。BigQueryに接続するための認証情報、Lambda、ネットワーク関連のパラメータを設定して、デプロイします。

Athena Google BigQuery Connectorは、Lambda環境変数を介して構成オプションを設定します。使用可能なパラメーターの詳細については、以下を参照してください。

パラメータ名	値の例	説明
Application name	AthenaGoogleBigQueryConnector	LambdaのApplication名に利用されます。
SpillBucket	my_bucket	Lambda関数から返されたデータがLambdaの制限を超えた場合、Athenaが超過分を読み取るためにデータを書き込むバケットです。
SpillPrefix	temporary/split	（オプション）デフォルトは、バケット内の 'athena-federation-spill' というサブフォルダです。spill_bucketと一緒に使用され、これは上記のバケット内で大きな応答がspillされるパスです。この場所にS3ライフサイクルを設定して、X日/時間後に古いspillを削除する必要があります。
ConcurrencyLimit	10	デフォルトでは、10です。この値の上限は、Google プロジェクトで設定した Google Bigquery Quota の制限に依存します。
DisableSpillEncryption	True or False	（オプション）デフォルトはFalseであるため、S3にスピルされたデータは、ランダムに生成されたキーを使用してAES-GMCを使用するか、KMSを使用してキーを生成して暗号化されます。これをfalseに設定すると、スピル暗号化が無効になります。特にS3の流出場所でS3サーバー側暗号化を使用している場合は、パフォーマンスを向上させるためにこれを無効にすることをお勧めします。
GCPProjectID	semiotic-primer-1234567	このコネクタが読み取る必要のあるデータセットを含むプロジェクトID（プロジェクト名ではない）。
LambdaFunctionName	athena-federated-query	LambdaのFunction名です。
LambdaFunctionRole	arn:aws:iam::123456789012:role/bigquery-afq	Lambda関数の実行ロールをARNで指定します。
LambdaMemory	3008	Lambdaのメモリサイズを指定します。（128〜3008MB）
LambdaTimeout	900	Lambdaの実行タイムアウトを指定します。（1〜900秒）
SecretManagerGCPCredsName	arn:aws:secretsmanager:ap-northeast-1:123456789012:secret:dev/devio/bigquery-abc123	Google Cloudのサービスアカウントの秘密鍵を保存したsecretのARNを指定します。
SecurityGroupIds	sg-1234567890	Lambda 関数のセキュリティグループIDを指定します。複数指定する場合は、カンマでつなげてください。
SubnetIds	subnet-1234567890	Lambda 関数をデプロイするサブネットを指定します。複数指定する場合は、カンマでつなげてください。

※ spillとは、メモリ内に保存しきれないデータを共有ストレージ（この場合はS3）に一時的に退避することを表します。例えば、spill dataは「あふれたデータ」といった意味でとらえてください。

デプロイが開始すると以下の画面に遷移し、画面下にLogical IDのリンクが表示されます。

データソースコネクタの実態は、Lambda Functionです。Logical IDをクリックすると、Lambda Functionの画面が表示されます。

デプロイ状況の確認

データソースコネクタは、CloudFormationでデプロイされますので、デプロイの状況は、CloudFormationの画面から確認できます。なお、データソースコネクタの削除（アンデプロイ）は、このCloudFormationのスタックを削除します。

テーブル名の指定方法

クエリしたい対象のテーブルの指定は、プレフィックスlambda:+ Lambda関数名、BigQueryのデータセット名（スキーマ名）、BigQueryのテーブル名を指定します。プレフィックスlambda:+ Lambda関数名については、ダブルクオートで括ることを忘れないでください。

例えば、BigQueryからcm-ishikawa-satoru.example.earthquake10と指定する場合、Athenaのfederated Queryでは、"lambda:athena-federated-query".example.earthquake10のように指定します。

クエリの実行

デプロイしたデータソースコネクタでクエリした結果は、以下のとおりです。BigQueryのクエリエディタで表示されたデータが、Amazon Athenaでも同じように表示されることを確認できました。

Partitionと分割について

BigQueryのレコードの取得は、concurrencyLimitに指定したページ数に分割され、クエリの実行中に制限値とオフセットを使用します。BigQueryのパーティション値に基づいて分割する仕様ではありません。

以下、2022/04/24時点のawslabs/aws-athena-query-federation/athena-google-bigqueryのPartitions and Splitsより、引用。

Currently splits are not based on partitions and based on configured concurrencyLimit environment variable. It will decide the page count for split and uses the limit and offset while executing the query instead of using partition value in the query. For predicate push down queries no splits are being used since it returns the lesser results. In addition,splits strategy is applicable only for non predicate push down queries(for select * queries). We can further increase the concurrencyLimit as per Google Bigquery Quota limits configured within Google project.

現在、分割はパーティションによらず、設定された concurrencyLimit 環境変数に基づき行われます。分割のためのページ数を決定し、クエリでパーティション値を使用する代わりに、クエリの実行中に制限値とオフセットを使用します。述語のプッシュダウンクエリでは、より少ない結果を返すため、分割は使用されません。また、分割戦略は、述語のないプッシュダウンクエリ（select * クエリ）にのみ適用されます。concurrencyLimit は、Google プロジェクトで設定した Google Bigquery Quota の制限に従ってさらに増やすことができます。

最後に

Athenaのfederated Queryのデータソースコネクタを用いると簡単に、BigQueryのデータにアクセスできることが確認できました。BigQueryライブデータとAthenaのデータをアドホックにブレンディングして分析できるようになります。データソースコネクタのデプロイの際に、Google CloudのプロジェクトIDを指定するだけで、プロジェクト配下にある全てのBigQueryのデータセット/テーブルにアクセスできるようになります。AthenaやGlue Data Catalogなどで、BigQueryのメタデータ管理することなく、最新のデータにアクセスできることは、運用面・保守面でもメリットが大きいはずです。このデータソースコネクタは、よく考え抜かれて設計・実装されていると感心しました。

BigQueryのデータソースコネクタを効果的に活用するには、Lambdaの実行時間（最大900秒）という制限と、LambdaがBigQueryのパーティション毎に並列分散で実行してデータを取得するという仕様を考慮したほうが良いでしょう。また、AthenaからBigQueryへのアクセスは、VPC Lambdaが仲介するため、20秒程度のオーバヘッドが生じることも忘れてはなりません。ユースケースに合わせて、Lambdaの並列数（ConcurrencyLimit）やLambdaのスペック（LambdaMemory）などのチューニング、これらのスケールアップ戦略・スケールアウト戦略を組み合わせても、Lambdaの実行時間の制限（最大900秒）を超えてしまう場合には、実行するクエリ自体のチューニングを組み合わせると良いでしょう。

データソースコネクタは、オープンソースで公開していますので、仕様の詳細を確認したり、ユースケースに合わせて実装を見直し、デプロイするということも可能です。

参考文献